package cn.doitedu.demo4;

import cn.doitedu.beans.EventBean;
import cn.doitedu.beans.RuleMeta;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/5/14
 * @Desc: 学大数据，上多易教育
 **/
@Slf4j
public class Entrypoint {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);

        // 构造kafkaSource
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("doitedu:9092")
                .setTopics("ods-user-action-log")
                .setGroupId("ggg")
                .setClientIdPrefix("client-001")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .setStartingOffsets(OffsetsInitializer.latest())
                .build();

        DataStreamSource<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "kafka-source");

        SingleOutputStreamOperator<EventBean> beanStream = stream.map(json -> JSON.parseObject(json, EventBean.class));

        KeyedStream<EventBean, Long> keyedUserActions = beanStream.keyBy(EventBean::getUser_id);


        /**
         * 实时监控 规则元数据表
         */
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("doitedu")
                .port(3306)
                .databaseList("doit47") // set captured database, If you need to synchronize the whole database, Please set tableList to ".*".
                .tableList("doit47.rule_meta") // set captured table
                .username("root")
                .password("root")
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .build();


        DataStreamSource<String> ruleMetaCdc = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "cdc");

        // 解析cdc读过来的json，封装成： 规则元数据bean
        SingleOutputStreamOperator<RuleMeta> ruleMetaStream = ruleMetaCdc.map(json -> {

            JSONObject jsonObject = JSON.parseObject(json);
            // 取出操作类型
            String op = jsonObject.getString("op");

            JSONObject dataObject;
            if (op.equals("d")) {
                dataObject = jsonObject.getJSONObject("before");
            } else {
                dataObject = jsonObject.getJSONObject("after");
            }

            // 往数据object中填充op字段
            dataObject.put("op", op);

            // 再把数据object转成我们自己的RuleMeta Bean
            return JSON.parseObject(dataObject.toJSONString(), RuleMeta.class);

        });

        // 广播规则元数据流
        MapStateDescriptor<String, RuleMeta> desc = new MapStateDescriptor<>("meta_bc_state", String.class, RuleMeta.class);
        BroadcastStream<RuleMeta> ruleMetaBroadcastStream = ruleMetaStream.broadcast(desc);


        // 将 用户行为流  连接   规则元数据广播流
        BroadcastConnectedStream<EventBean, RuleMeta> connectedStream = keyedUserActions.connect(ruleMetaBroadcastStream);


        // 对 连接好的两个流，进行统一处理
        SingleOutputStreamOperator<String> resultStream = connectedStream.process(new KeyedBroadcastProcessFunction<Long, EventBean, RuleMeta, String>() {


            HashMap<String, RuleCalculator> ruleCalculatorPool = new HashMap<>();


            /**
             *  处理 连接流中的  行为流（keyed流）
             */
            @Override
            public void processElement(EventBean eventBean, KeyedBroadcastProcessFunction<Long, EventBean, RuleMeta, String>.ReadOnlyContext ctx, Collector<String> out) throws Exception {

                for (Map.Entry<String, RuleCalculator> entry : ruleCalculatorPool.entrySet()) {
                    RuleCalculator ruleCalculator = entry.getValue();

                    ruleCalculator.calc(eventBean,out);
                }

            }


            /**
             *  处理 连接流中的  规则元数据流（广播流）
             */
            @Override
            public void processBroadcastElement(RuleMeta ruleMeta, KeyedBroadcastProcessFunction<Long, EventBean, RuleMeta, String>.Context ctx, Collector<String> out) throws Exception {

                String ruleId = ruleMeta.getRule_id();
                String op = ruleMeta.getOp();
                int status = ruleMeta.getStatus();

                // 根据收到的规则元数据操作类型，来对系统的  “运算机池” 做对应的操作
                //  op = c / r / u  且 status = 1    ==>  构造运算机对象，并用规则参数对它初始化，放入运算机池
                if ((op.equals("c") || op.equals("r") || op.equals("u")) && status == 1) {
                    // 构造规则运算机对象
                    RuleCalculator ruleCalculator = new RuleModel_01_Calculator();

                    // 用规则元数据中的规则参数，来初始化运算机对象
                    ruleCalculator.init(getRuntimeContext(), ruleMeta.getRule_param_json());

                    // 将初始化好的运算机对象，放入运算机池
                    ruleCalculatorPool.put(ruleId, ruleCalculator);

                    log.info("上线或更新了一个规则运算机: {}",ruleId);

                }
                //  op = d  或   op=u && status = 0  ==> 从运算机池把该规则的运算机对象移除
                else if ((op.equals("d") || (op.equals("u") && status == 0))) {
                    ruleCalculatorPool.remove(ruleId);
                    log.info("下线了一个规则运算机: {}",ruleId);

                } else {
                    log.error("收到不支持的规则元数据操作类型：" + op);
                }
            }
        });


        resultStream.print();

        env.execute("i love you all");


    }


}
